package com.bee.plus.ai.moonshot;

import com.bee.plus.ai.MessagePrepare;
import com.bee.plus.ai.baidu.BaiduAiServiceImpl;
import com.bee.plus.ai.baidu.HttpChatClient;
import com.bee.plus.ai.baidu.ResponseCallback;
import com.bee.plus.dto.ContentReq;
import com.bee.plus.dto.MessageReq;
import com.bee.plus.dto.MessageResp;
import com.bee.plus.util.Conversation;
import com.bee.plus.util.ConversationThreadLocal;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;

/**
 * @author xupei
 */

@Service("kimiAi")
public class KimiAiServiceImpl extends MessagePrepare {

    private Logger logger= LoggerFactory.getLogger(KimiAiServiceImpl.class);

    @Autowired
    private KimiAiManager kimiManager;
    @Override
    protected MessageReq preprocessMessage(MessageReq messageReq) {
        Conversation conversation= ConversationThreadLocal.getCurrentConversation();
        List<ContentReq> list=storeManger.getDb().getContentHistoryByConversationId(conversation.getConversationId());
        if(list!=null&&list.size()!=0){
            list=list.subList(list.size()-2,list.size());
            list.add(messageReq.getMessages().get(0));
            messageReq.setMessages(list);
        }
        storeManger.getDb().storeChatInfo(messageReq.getMessages().get(messageReq.getMessages().size()-1),conversation.getConversationId());


        return messageReq;
    }

    @Override
    protected MessageResp processMessage(MessageReq messageReq) {
        return null;
    }

    @Override
    protected Flowable<MessageResp> processMessageStream(MessageReq messageReq) {
        messageReq.setModel(kimiManager.getModel());
        messageReq.setTemperature(0.3);
        Flowable<MessageResp> messageRespFlowable=  Flowable.create(emitter -> {
            List<ContentReq> messagesList = messageReq.getMessages();
            messageReq.setMessages(messagesList);
            messageReq.setStream(true);
            logger.info("kimi ai service req {}", messageReq.toString());
            Schedulers.newThread().scheduleDirect(()->{

                String urlString = kimiManager.getChatUrl();
                HttpChatClient client = new HttpChatClient(urlString);
                try{
                    client.sendMessageAndGetResponseKimi(kimiManager.getApiKey(), messageReq, new ResponseCallback() {
                        @Override
                        public void onResponse( MessageResp messageResp) {
                            // 处理响应逻辑
                            emitter.onNext(messageResp);
                        }

                        @Override
                        public void onError(Exception e) {
                            // 处理错误逻辑
                            emitter.onError(e);
                        }

                        @Override
                        public void finish(MessageResp messageResp) {
                            emitter.onComplete();
                        }
                    });
                }catch (IOException e){
                    e.printStackTrace();
                    emitter.onError(e);
                }


            });

        }, BackpressureStrategy.BUFFER);


        return messageRespFlowable;
    }

    @Override
    public void storeChatMessage(MessageResp messageResp,Long conversationId) {
        super.postprocessResponse(messageResp,conversationId);
    }
}
